package com.yykj.tencent.im.server.tdmq.producer;

import com.yykj.tencent.im.server.exception.ProducerInitException;
import com.yykj.tencent.im.server.utils.SchemaUtil;
import org.apache.pulsar.client.api.*;


import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;

/**
 * TDMQ 模板
 * @param <T> 泛型类型
 */
public class MessageTemplate<T> {
    private final PulsarClient pulsarClient;
    private final Map<String, Producer<T>> producers = new ConcurrentHashMap<>();

    public MessageTemplate(PulsarClient pulsarClient) {
        this.pulsarClient = pulsarClient;
    }

    /**
     * 同步
     * @param topic 主题
     * @param message  消息内容
     * @return
     * @throws PulsarClientException
     */
    public MessageId send(String topic, T message) throws PulsarClientException {
        Producer<T> producer = producers.computeIfAbsent(topic, key -> buildProducer(key, message.getClass()));
        return producer.send(message);
    }

    /**
     * 异步
     * @param topic 主题
     * @param message 消息内容
     * @return
     */
    public CompletableFuture<MessageId> sendAsync(String topic, T message) {
        Producer<T> producer = producers.computeIfAbsent(topic, key -> buildProducer(key, message.getClass()));
        return producer.sendAsync(message);
    }

    /**
     * 创建消息
     * @param topic
     * @param message
     * @return
     */
    public TypedMessageBuilder<T> createMessage(String topic, T message) {
        Producer<T> producer = producers.computeIfAbsent(topic, key -> buildProducer(key, message.getClass()));
        return producer.newMessage().value(message);
    }

    /**
     * 创建生产者
     * @param topic
     * @param holder
     * @return
     */
    private Producer<T> buildProducer(String topic, Class<?> holder) {
        try {
            @SuppressWarnings("unchecked")
            Class<T> tClass = (Class<T>) holder;
            return pulsarClient.newProducer(SchemaUtil.getSchema(tClass))
                    .messageRoutingMode(MessageRoutingMode.RoundRobinPartition)
                    .hashingScheme(HashingScheme.Murmur3_32Hash)
                    .topic(topic)
                    .create();
        } catch (PulsarClientException e) {
            throw new ProducerInitException("Failed to init producer.", e);
        }
    }
}
